package com.rp.demo.laqu;

import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;
import com.rp.demo.config.RabbitLaQuConfig;
import com.rp.demo.dto.MessageEntity;
import com.rp.demo.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Component
@Slf4j
public class LaQuConsumer {

    /**
     * 消费者
     *
     * @throws Exception
     */
    public void consumerLaQu() throws Exception {

        //创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("101.200.133.224");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");

        //创建连接
        Connection connection = factory.newConnection();
        //创建消息信道
        Channel channel = connection.createChannel();
        //存放消息实体List
        List<MessageEntity> list = new ArrayList<>();

        long start = System.currentTimeMillis();
        //MessageEntity entity =new MessageEntity();

        long tag = 0L;
        log.info("=====================================================");
        while (true) {
            //拉取消息
            GetResponse response = channel.basicGet(RabbitLaQuConfig.LAQU_QUEUE_NAME, false);
            if (response == null) {
                //间隔时间，如果超过10s还没有消费到新到消息，则将消息入库，保证实效性
                long interval = System.currentTimeMillis() - start;
                log.info("=======1====={}", interval);
                log.info("=======2====={}", "list中有" + list.size() + "条记录");
                if (CollectionUtils.isEmpty(list) && interval > 10000) {
                    log.info("队列中已经是空数据,停止消费");
                    break;
                }
                if (CollectionUtils.isNotEmpty(list) && interval > 10000) {
                    //批量确认消息
                    channel.basicAck(tag, true);
                    //模仿业务处理
                    log.info("处理业务逻辑:{}", list.size());
                    list.forEach(item -> log.info(item.getMessage()));
                    list.clear();

                    Thread.sleep(1000);
                    start = System.currentTimeMillis();
                }
                Thread.sleep(500);
                continue;
            }
            String str = new String(response.getBody());
            MessageEntity messageEntity = JsonUtil.strToObj(str, MessageEntity.class);
            //entity.setMessage(str);
            //entity.setTag(response.getEnvelope().getDeliveryTag());
            list.add(messageEntity);
            tag = response.getEnvelope().getDeliveryTag();
            //100条消息批量入库一次
            if (list.size() % 10 == 0) {
                log.info("===========3====={}", list.size() / 10);
                //批量确认消息
                //channel.basicAck(entity.getTag(),true);
                channel.basicAck(tag, true);
                log.info("===========处理业务==========={}", list.size());
                list.forEach(item -> log.info(item.getMessage()));
                list.clear();
                //模仿业务处理
                Thread.sleep(1000);
                start = System.currentTimeMillis();
            }
        }
    }

}
